/*-------------------------------------------------------------------------
 *
 * hashutil.c
 *	  Postgres哈希实现的实用程序代码。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/hash/hashutil.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/hash.h"
#include "access/reloptions.h"
#include "access/relscan.h"
#include "port/pg_bitutils.h"
#include "storage/buf_internals.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"

#define CALC_NEW_BUCKET(old_bucket, lowmask) \
			old_bucket | (lowmask + 1)

/*
 * _hash_checkqual -- 索引元组是否满足扫描条件？
 */
bool _hash_checkqual(IndexScanDesc fc_scan, IndexTuple fc_itup)
{
	/*
	 * 目前，我们无法检查任何扫描条件，因为我们没有
	 * 原始索引条目值来提供给 sk_func。始终
	 * 返回 true；我们期望 hashgettuple 已经设置了重新检查标志
	 * 以使主索引扫描代码执行该操作。
	 */
#ifdef NOT_USED
	TupleDesc	fc_tupdesc = RelationGetDescr(fc_scan->indexRelation);
	ScanKey		fc_key = fc_scan->keyData;
	int			fc_scanKeySize = fc_scan->numberOfKeys;

	while (fc_scanKeySize > 0)
	{
		Datum		fc_datum;
		bool		fc_isNull;
		Datum		fc_test;

		fc_datum = index_getattr(fc_itup,
							  fc_key->sk_attno,
							  fc_tupdesc,
							  &fc_isNull);

		/* 假设 sk_func 是严格的 */
		if (fc_isNull)
			return false;
		if (fc_key->sk_flags & SK_ISNULL)
			return false;

		fc_test = FunctionCall2Coll(&fc_key->sk_func, fc_key->sk_collation,
								 fc_datum, fc_key->sk_argument);

		if (!DatumGetBool(fc_test))
			return false;

		fc_key++;
		fc_scanKeySize--;
	}
#endif

	return true;
}

/*
 * _hash_datum2hashkey -- 给定一个 Datum，调用索引的哈希函数
 *
 * 假设 Datum 是索引的列类型，因此我们可以使用
 * 通用索引代码为我们跟踪的 "主" 哈希函数。
 */
uint32 _hash_datum2hashkey(Relation fc_rel, Datum fc_key)
{
	FmgrInfo   *fc_procinfo;
	Oid			fc_collation;

	/* XXX 假设索引只有一个属性 */
	fc_procinfo = index_getprocinfo(fc_rel, 1, HASHSTANDARD_PROC);
	fc_collation = fc_rel->rd_indcollation[0];

	return DatumGetUInt32(FunctionCall1Coll(fc_procinfo, fc_collation, fc_key));
}

/*
 * _hash_datum2hashkey_type -- 给定一个指定类型的 Datum，
 *		 以与此索引兼容的方式进行哈希
 *
 * 这比 _hash_datum2hashkey 要昂贵得多，因此仅在
 * 跨类型情况下使用它。
 */
uint32 _hash_datum2hashkey_type(Relation fc_rel, Datum fc_key, Oid fc_keytype)
{
	RegProcedure fc_hash_proc;
	Oid			fc_collation;

	/* XXX 假设索引只有一个属性 */
	fc_hash_proc = get_opfamily_proc(fc_rel->rd_opfamily[0],
								  fc_keytype,
								  fc_keytype,
								  HASHSTANDARD_PROC);
	if (!RegProcedureIsValid(fc_hash_proc))
		elog(ERROR, "missing support function %d(%u,%u) for index \"%s\"",
			 HASHSTANDARD_PROC, fc_keytype, fc_keytype,
			 RelationGetRelationName(fc_rel));
	fc_collation = fc_rel->rd_indcollation[0];

	return DatumGetUInt32(OidFunctionCall1Coll(fc_hash_proc, fc_collation, fc_key));
}

/*
 * _hash_hashkey2bucket -- 确定哈希键映射到哪个桶。
 */
Bucket
_hash_hashkey2bucket(uint32 fc_hashkey, uint32 fc_maxbucket,
					 uint32 fc_highmask, uint32 fc_lowmask)
{
	Bucket		fc_bucket;

	fc_bucket = fc_hashkey & fc_highmask;
	if (fc_bucket > fc_maxbucket)
		fc_bucket = fc_bucket & fc_lowmask;

	return fc_bucket;
}

/*
 * _hash_spareindex -- 返回备用索引 / 全局拆分点阶段的
 *					   桶
 */
uint32 _hash_spareindex(uint32 fc_num_bucket)
{
	uint32		fc_splitpoint_group;
	uint32		fc_splitpoint_phases;

	fc_splitpoint_group = pg_ceil_log2_32(fc_num_bucket);

	if (fc_splitpoint_group < HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE)
		return fc_splitpoint_group;

	/* 考虑单阶段组 */
	fc_splitpoint_phases = HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE;

	/* 在拆分点组之前考虑多阶段组 */
	fc_splitpoint_phases +=
		((fc_splitpoint_group - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) <<
		 HASH_SPLITPOINT_PHASE_BITS);

	/* 考虑当前组内的阶段 */
	fc_splitpoint_phases +=
		(((fc_num_bucket - 1) >>
		  (fc_splitpoint_group - (HASH_SPLITPOINT_PHASE_BITS + 1))) &
		 HASH_SPLITPOINT_PHASE_MASK);	/* 转换为 0 基值。 */

	return fc_splitpoint_phases;
}

/*
 *	_hash_get_totalbuckets -- 返回直到给定拆分点阶段
 *							分配的总桶数。
 */
uint32 _hash_get_totalbuckets(uint32 fc_splitpoint_phase)
{
	uint32		fc_splitpoint_group;
	uint32		fc_total_buckets;
	uint32		fc_phases_within_splitpoint_group;

	if (fc_splitpoint_phase < HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE)
		return (1 << fc_splitpoint_phase);

	/* 获取拆分点的组 */
	fc_splitpoint_group = HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE;
	fc_splitpoint_group +=
		((fc_splitpoint_phase - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) >>
		 HASH_SPLITPOINT_PHASE_BITS);

	/* 考虑拆分点组之前的桶 */
	fc_total_buckets = (1 << (fc_splitpoint_group - 1));

	/* 考虑拆分点组内的桶 */
	fc_phases_within_splitpoint_group =
		(((fc_splitpoint_phase - HASH_SPLITPOINT_GROUPS_WITH_ONE_PHASE) &
		  HASH_SPLITPOINT_PHASE_MASK) + 1); /* 从 0 基值转换为 1 基值 */
	fc_total_buckets +=
		(((1 << (fc_splitpoint_group - 1)) >> HASH_SPLITPOINT_PHASE_BITS) *
		 fc_phases_within_splitpoint_group);

	return fc_total_buckets;
}

/*
 * _hash_checkpage -- 对所有哈希页格式进行一致性检查
 *
 * 如果 flags 不是零，则它是可接受页面类型的按位或
 * （hasho_flag & LH_PAGE_TYPE 的值）。
 */
void _hash_checkpage(Relation fc_rel, Buffer fc_buf, int fc_flags)
{
	Page		fc_page = BufferGetPage(fc_buf);

	/*
	 * ReadBuffer验证每个新读取的页面是否通过
	 * PageHeaderIsValid，这意味着它要么包含一个相当正常的
	 * 页面头，要么全为零。然而，我们必须防范全零
	 * 的情况。
	 */
	if (PageIsNew(fc_page))
		ereport(ERROR,
				(errcode(ERRCODE_INDEX_CORRUPTED),
				 errmsg("index \"%s\" contains unexpected zero page at block %u",
						RelationGetRelationName(fc_rel),
						BufferGetBlockNumber(fc_buf)),
				 errhint("Please REINDEX it.")));

	/*
	 * 另外检查特殊区域是否正常。
	 */
	if (PageGetSpecialSize(fc_page) != MAXALIGN(sizeof(HashPageOpaqueData)))
		ereport(ERROR,
				(errcode(ERRCODE_INDEX_CORRUPTED),
				 errmsg("index \"%s\" contains corrupted page at block %u",
						RelationGetRelationName(fc_rel),
						BufferGetBlockNumber(fc_buf)),
				 errhint("Please REINDEX it.")));

	if (fc_flags)
	{
		HashPageOpaque fc_opaque = HashPageGetOpaque(fc_page);

		if ((fc_opaque->hasho_flag & fc_flags) == 0)
			ereport(ERROR,
					(errcode(ERRCODE_INDEX_CORRUPTED),
					 errmsg("index \"%s\" contains corrupted page at block %u",
							RelationGetRelationName(fc_rel),
							BufferGetBlockNumber(fc_buf)),
					 errhint("Please REINDEX it.")));
	}

	/*
	 * 在检查元页面时，还要验证魔法数字和版本。
	 */
	if (fc_flags == LH_META_PAGE)
	{
		HashMetaPage fc_metap = HashPageGetMeta(fc_page);

		if (fc_metap->hashm_magic != HASH_MAGIC)
			ereport(ERROR,
					(errcode(ERRCODE_INDEX_CORRUPTED),
					 errmsg("index \"%s\" is not a hash index",
							RelationGetRelationName(fc_rel))));

		if (fc_metap->hashm_version != HASH_VERSION)
			ereport(ERROR,
					(errcode(ERRCODE_INDEX_CORRUPTED),
					 errmsg("index \"%s\" has wrong hash version",
							RelationGetRelationName(fc_rel)),
					 errhint("Please REINDEX it.")));
	}
}

bytea * hashoptions(Datum fc_reloptions, bool fc_validate)
{
	static const relopt_parse_elt fc_tab[] = {
		{"fillfactor", RELOPT_TYPE_INT, offsetof(HashOptions, fillfactor)},
	};

	return (bytea *) build_reloptions(fc_reloptions, fc_validate,
									  RELOPT_KIND_HASH,
									  sizeof(HashOptions),
									  fc_tab, lengthof(fc_tab));
}

/*
 * _hash_get_indextuple_hashkey - 获取哈希索引元组的哈希键值
 */
uint32 _hash_get_indextuple_hashkey(IndexTuple fc_itup)
{
	char	   *fc_attp;

	/*
	 * 我们假设哈希键是第一个属性且不能为空，因此
	 * 这可以粗略地但非常非常便宜地完成 ...
	 */
	fc_attp = (char *) fc_itup + IndexInfoFindDataOffset(fc_itup->t_info);
	return *((uint32 *) fc_attp);
}

/*
 * _hash_convert_tuple - 将原始索引数据转换为哈希键
 *
 * 输入：用户数据列的值数组和isnull数组
 * 输出：适合传递给index_form_tuple()的索引元组的值数组和isnull数组
 *
 * 如果成功则返回true，如果未成功（因为存在null值）则返回false。
 * 对于false结果，给定数据不需要被索引。
 *
 * 注意：调用者知道索引列数组的长度始终为1。
 * 理论上，可能有多个输入列，尽管我们目前不支持这一点。
 */
bool _hash_convert_tuple(Relation fc_index,
					Datum *fc_user_values, bool *fc_user_isnull,
					Datum *fc_index_values, bool *fc_index_isnull)
{
	uint32		fc_hashkey;

	/*
	 * 我们不将null值插入哈希索引。这是可以的，因为
	 * 唯一支持的搜索操作符是'='，我们假设它是严格的。
	 */
	if (fc_user_isnull[0])
		return false;

	fc_hashkey = _hash_datum2hashkey(fc_index, fc_user_values[0]);
	fc_index_values[0] = UInt32GetDatum(fc_hashkey);
	fc_index_isnull[0] = false;
	return true;
}

/*
 * _hash_binsearch - 返回在页面中指定哈希值应被搜索或插入的位置的偏移量。
 *
 * 我们使用二分查找，依赖于现有条目按哈希键排序的假设。
 *
 * 返回第一个索引条目的偏移量，其hashkey >= hash_value，
 * 或者如果hash_value大于页面中的所有现有哈希键，则返回页面的最大偏移加一。
 * 这是开始搜索或插入新项的适当位置。
 */
OffsetNumber _hash_binsearch(Page fc_page, uint32 fc_hash_value)
{
	OffsetNumber fc_upper;
	OffsetNumber fc_lower;

	/* 循环不变式： lower <= desired place <= upper */
	fc_upper = PageGetMaxOffsetNumber(fc_page) + 1;
	fc_lower = FirstOffsetNumber;

	while (fc_upper > fc_lower)
	{
		OffsetNumber fc_off;
		IndexTuple	fc_itup;
		uint32		fc_hashkey;

		fc_off = (fc_upper + fc_lower) / 2;
		Assert(OffsetNumberIsValid(fc_off));

		fc_itup = (IndexTuple) PageGetItem(fc_page, PageGetItemId(fc_page, fc_off));
		fc_hashkey = _hash_get_indextuple_hashkey(fc_itup);
		if (fc_hashkey < fc_hash_value)
			fc_lower = fc_off + 1;
		else
			fc_upper = fc_off;
	}

	return fc_lower;
}

/*
 * _hash_binsearch_last
 *
 * 与上述相同，除了如果页面中有多个匹配项，
 * 我们返回最后一个的偏移量而不是第一个，
 * 并且可能的输出范围是0..maxoffset，而不是1..maxoffset+1。
 * 这在反向扫描中启动新页面时非常方便。
 */
OffsetNumber _hash_binsearch_last(Page fc_page, uint32 fc_hash_value)
{
	OffsetNumber fc_upper;
	OffsetNumber fc_lower;

	/* 循环不变式： lower <= desired place <= upper */
	fc_upper = PageGetMaxOffsetNumber(fc_page);
	fc_lower = FirstOffsetNumber - 1;

	while (fc_upper > fc_lower)
	{
		IndexTuple	fc_itup;
		OffsetNumber fc_off;
		uint32		fc_hashkey;

		fc_off = (fc_upper + fc_lower + 1) / 2;
		Assert(OffsetNumberIsValid(fc_off));

		fc_itup = (IndexTuple) PageGetItem(fc_page, PageGetItemId(fc_page, fc_off));
		fc_hashkey = _hash_get_indextuple_hashkey(fc_itup);
		if (fc_hashkey > fc_hash_value)
			fc_upper = fc_off - 1;
		else
			fc_lower = fc_off;
	}

	return fc_lower;
}

/*
 *	_hash_get_oldblock_from_newbucket() -- 从当前（新）桶被拆分的桶中获取块号。
 */
BlockNumber _hash_get_oldblock_from_newbucket(Relation fc_rel, Bucket fc_new_bucket)
{
	Bucket		fc_old_bucket;
	uint32		fc_mask;
	Buffer		fc_metabuf;
	HashMetaPage fc_metap;
	BlockNumber fc_blkno;

	/*
	 * 要从当前桶获取旧桶，我们需要一个掩码以对表的下半部分取模。
	 * 这个掩码存储在元页面中，作为hashm_lowmask，但在这里我们无法依赖相同的值，
	 * 因为我们需要一个在桶拆分开始时盛行的lowmask值。
	 * 对新桶的最高有效位进行掩码将给我们旧桶。
	 */
	fc_mask = (((uint32) 1) << (fls(fc_new_bucket) - 1)) - 1;
	fc_old_bucket = fc_new_bucket & fc_mask;

	fc_metabuf = _hash_getbuf(fc_rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	fc_blkno = BUCKET_TO_BLKNO(fc_metap, fc_old_bucket);

	_hash_relbuf(fc_rel, fc_metabuf);

	return fc_blkno;
}

/*
 *	_hash_get_newblock_from_oldbucket() -- 获取将在从旧桶拆分后生成的桶的块号。
 *
 * 这用于根据当前表的下半部分从旧桶查找新桶。
 * 这主要用于完成不完整的拆分，在这我们确定不会有超过一个桶正在从旧桶拆分。
 */
BlockNumber _hash_get_newblock_from_oldbucket(Relation fc_rel, Bucket fc_old_bucket)
{
	Bucket		fc_new_bucket;
	Buffer		fc_metabuf;
	HashMetaPage fc_metap;
	BlockNumber fc_blkno;

	fc_metabuf = _hash_getbuf(fc_rel, HASH_METAPAGE, HASH_READ, LH_META_PAGE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	fc_new_bucket = _hash_get_newbucket_from_oldbucket(fc_rel, fc_old_bucket,
													fc_metap->hashm_lowmask,
													fc_metap->hashm_maxbucket);
	fc_blkno = BUCKET_TO_BLKNO(fc_metap, fc_new_bucket);

	_hash_relbuf(fc_rel, fc_metabuf);

	return fc_blkno;
}

/*
 * _hash_get_newbucket_from_oldbucket() -- 从当前（旧）桶获取将在分裂后生成的新桶。
 *
 * 这用于从旧桶中找到新桶。通过将旧桶与当前表半部分的最高有效位进行或运算，可以获得新桶（传递给此函数的lowmask可以用来识别当前表半部分的msb）。可能会有多个桶是从当前桶分裂出来的。我们需要第一个存在的这样的桶。
 * 调用者必须确保从旧桶中最多只发生了一次分裂。
 */
Bucket
_hash_get_newbucket_from_oldbucket(Relation fc_rel, Bucket fc_old_bucket,
								   uint32 fc_lowmask, uint32 fc_maxbucket)
{
	Bucket		fc_new_bucket;

	fc_new_bucket = CALC_NEW_BUCKET(fc_old_bucket, fc_lowmask);
	if (fc_new_bucket > fc_maxbucket)
	{
		fc_lowmask = fc_lowmask >> 1;
		fc_new_bucket = CALC_NEW_BUCKET(fc_old_bucket, fc_lowmask);
	}

	return fc_new_bucket;
}

/*
 * _hash_kill_items - 为索引扫描调用者告知已被杀死的项目设置LP_DEAD状态。
 *
 * scan->opaque，通过so本地引用，包含有关当前页面和上面已被杀死的元组的信息（通常，这应仅在so->numKilled > 0时调用）。
 *
 * 调用者对页面没有锁，并且可能会或可能不会在缓冲区中固定该页面。请注意，读锁对于设置LP_DEAD状态（这仅仅是一个提示）是足够的。
 *
 * 调用者必须对桶缓冲区有固定，但可能会或可能不会对溢出缓冲区有固定，这由HashScanPosIsPinned(so->currPos)指示。
 *
 * 我们通过堆TID匹配项目，然后假设它们是要删除的正确项目。
 *
 * 在VACUUM开始时，桶中从未有任何活动扫描，因为VACUUM在主桶页面上获得了清理锁，而扫描保持一个固定。在VACUUM离开主桶页面后，但在完成整个桶之前，可以开始扫描，但它永远不会超过VACUUM，因为VACUUM总是在释放前一个页面的锁之前锁定下一个页面。因此，我们不必担心意外杀死已被用于无关元组的TID。
 */
void _hash_kill_items(IndexScanDesc fc_scan)
{
	HashScanOpaque fc_so = (HashScanOpaque) fc_scan->opaque;
	Relation	fc_rel = fc_scan->indexRelation;
	BlockNumber fc_blkno;
	Buffer		fc_buf;
	Page		fc_page;
	HashPageOpaque fc_opaque;
	OffsetNumber fc_offnum,
				fc_maxoff;
	int			fc_numKilled = fc_so->numKilled;
	int			fc_i;
	bool		fc_killedsomething = false;
	bool		fc_havePin = false;

	Assert(fc_so->numKilled > 0);
	Assert(fc_so->killedItems != NULL);
	Assert(HashScanPosIsValid(fc_so->currPos));

	/*
	 * 始终重置扫描状态，以便我们不在其他页面上查找相同的项目。
	 */
	fc_so->numKilled = 0;

	fc_blkno = fc_so->currPos.currPage;
	if (HashScanPosIsPinned(fc_so->currPos))
	{
		/*
		 * 我们已经对该缓冲区进行了固定，因此，我们需要做的就是对其获取锁。
		 */
		fc_havePin = true;
		fc_buf = fc_so->currPos.buf;
		LockBuffer(fc_buf, BUFFER_LOCK_SHARE);
	}
	else
		fc_buf = _hash_getbuf(fc_rel, fc_blkno, HASH_READ, LH_OVERFLOW_PAGE);

	fc_page = BufferGetPage(fc_buf);
	fc_opaque = HashPageGetOpaque(fc_page);
	fc_maxoff = PageGetMaxOffsetNumber(fc_page);

	for (fc_i = 0; fc_i < fc_numKilled; fc_i++)
	{
		int			fc_itemIndex = fc_so->killedItems[fc_i];
		HashScanPosItem *fc_currItem = &fc_so->currPos.items[fc_itemIndex];

		fc_offnum = fc_currItem->indexOffset;

		Assert(fc_itemIndex >= fc_so->currPos.firstItem &&
			   fc_itemIndex <= fc_so->currPos.lastItem);

		while (fc_offnum <= fc_maxoff)
		{
			ItemId		fc_iid = PageGetItemId(fc_page, fc_offnum);
			IndexTuple	fc_ituple = (IndexTuple) PageGetItem(fc_page, fc_iid);

			if (ItemPointerEquals(&fc_ituple->t_tid, &fc_currItem->heapTid))
			{
				/* 找到该项目 */
				ItemIdMarkDead(fc_iid);
				fc_killedsomething = true;
				break;			/* 退出内部搜索循环 */
			}
			fc_offnum = OffsetNumberNext(fc_offnum);
		}
	}

	/*
	 * 由于如果需要可以在稍后重新执行，因此标记为脏提示。每当我们标记任何内容为LP_DEAD时，我们还会设置页面的
	 * LH_PAGE_HAS_DEAD_TUPLES标志，这同样只是一个提示。
	 */
	if (fc_killedsomething)
	{
		fc_opaque->hasho_flag |= LH_PAGE_HAS_DEAD_TUPLES;
		MarkBufferDirtyHint(fc_buf, true);
	}

	if (fc_so->hashso_bucket_buf == fc_so->currPos.buf ||
		fc_havePin)
		LockBuffer(fc_so->currPos.buf, BUFFER_LOCK_UNLOCK);
	else
		_hash_relbuf(fc_rel, fc_buf);
}
